package com.hkbigdata.userprofile.app

import com.hkbigdata.userprofile.common.bean.TagInfo
import com.hkbigdata.userprofile.common.constcode.ConstCode
import com.hkbigdata.userprofile.common.dao.TagInfoDAO
import com.hkbigdata.userprofile.common.util.{MyClickHouseUtil, MyPropertiesUtil}
import org.apache.spark.SparkConf
import org.apache.spark.sql.SparkSession

import java.util.Properties
import scala.collection.mutable.ListBuffer

/**
 * @author Clown
 */
object TaskCkToBitmap {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "hkbigdata")
    val conf: SparkConf = new SparkConf().setAppName("tag_bitmap_app")
//      .setMaster("local[*]")
    val sparkSession: SparkSession = SparkSession.builder().config(conf).enableHiveSupport().getOrCreate()
    val properties: Properties = MyPropertiesUtil.load("config.properties")

    val taskId: String = args(0)
    val taskDate: String = args(1)

//    val tableName = "up_tag_merge_" + taskDate.replace("-", "")

//    val dbNameCH = properties.getProperty("user-profile.dbname.ch")
//    val clickhouseUrl = properties.getProperty("clickhouse.url")

    val tagInfoList: List[TagInfo] = TagInfoDAO.getTaskTagMergeOnTask()

    val stringTagList = new ListBuffer[TagInfo]
    val decimalTagList = new ListBuffer[TagInfo]
    val longTagList = new ListBuffer[TagInfo]
    val dateTagList = new ListBuffer[TagInfo]

    for (tagInfo <- tagInfoList) {
      tagInfo.tagValueType match {
        case ConstCode.TAG_VALUE_TYPE_STRING => stringTagList.append(tagInfo)
        case ConstCode.TAG_VALUE_TYPE_DECIMAL => decimalTagList.append(tagInfo)
        case ConstCode.TAG_VALUE_TYPE_LONG => longTagList.append(tagInfo)
        case ConstCode.TAG_VALUE_TYPE_DATE => dateTagList.append(tagInfo)
      }
    }

    //不同类型的数据进入不同的表
    insertTags("user_tag_value_string", stringTagList, taskDate)
    insertTags("user_tag_value_string", decimalTagList, taskDate)


  }

  // 列转行
  //    insert into user_tag_value_string
  //    select tag_code_value.1 as tag_code,
  //    tag_code_value.2 as tag_value ,
  //    groupBitmapState( uid ) us
  //      from (
  //        SELECT
  //          arrayJoin([('agegroup', agegroup),
  //          ('gender', gender),
  //          ('favor', favor)]) AS tag_code_value,
  //    uid
  //    FROM up_user_tag_merge
  //    )tv
  //    group by tag_code_value.1,tag_code_value.2
  def insertTags(tableName: String, tagList: ListBuffer[TagInfo],  taskDate: String): Unit = {
    if (tagList.nonEmpty) {
      val deleteSql = s" ALTER TABLE $tableName DELETE WHERE dt='${taskDate.replace("-", "")}'"
      println(deleteSql)
      MyClickHouseUtil.executeSql(deleteSql)

      val tagCodeValueSql: String = tagList.map { tagInfo: TagInfo => " ('" + tagInfo.tagCode + "'," + tagInfo.tagCode.toLowerCase + " ) " }.mkString(",")

      val selectSql: String =
        s"""|       select tag_code_value.1  tag_code,
            |        tag_code_value.2 tag_value,
            |        groupBitmapState(cast (uid as UInt64))  us ,'$taskDate' as dt
            |       from
            |       (
            |          select  arrayJoin( [$tagCodeValueSql] ) tag_code_value ,
            |          uid  from user_tag_merge_${taskDate.replace("-", "")}
            |       ) as tag
            |        where tag_code_value.2  <> ''
            |       group by
            |        tag_code_value.1,
            |        tag_code_value.2
              """.stripMargin
      val insertSql: String = s"insert into  $tableName " + selectSql
      println(insertSql)

      MyClickHouseUtil.executeSql(insertSql)
    }


  }
}
